import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.Properties;
import java.util.Random;

public class StudentInfoProducer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.128:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        Random random = new Random();
        String[] genders = {"1", "0"}; // 1 for male, 0 for female
        String[] classes = {"201", "202", "203"};

        try {
            while (true) {
                for (int i = 0; i < 10; i++) {
                    //System.out.println("===========================");
                    StringBuilder data = new StringBuilder();


                    // Class number
                    String className = classes[random.nextInt(classes.length)];
                    data.append(className).append("\t");

                    // Name
                    String name = generateRandomName();
                    data.append(name).append("\t");

                    // Gender
                    String gender = genders[random.nextInt(genders.length)];
                    data.append(gender).append("\t");

                    // Birth date (random date between 2000 and 2020)
                    LocalDate birthDate = generateRandomDate(2000, 2010);
                    data.append(birthDate.format(DateTimeFormatter.ofPattern("yyyyMMdd"))).append("\t");

                    // Student ID
                    Long studentId = generateStudentId();
                    data.append(studentId).append("\t");

                    // Semester 学期
                    Integer semester = 1 + random.nextInt(8); // Semester between 1 and 8 // Semester between 1 and 8
                    data.append(semester).append("\t");

                    // Enrollment status 入学状态
                    String enrollmentStatus = random.nextBoolean() ? "L" : "A"; // Randomly choose "L" (Left/Graduated) or "A" (Active)
                    data.append(enrollmentStatus);

                    // Send the data to Kafka
                    kafkaProducer.send(new ProducerRecord<>("stuInfo", data.toString()));
                    System.out.println("Sent data: " + data.toString());

                    // Wait for 5 seconds before sending the next batch
                }
                    Thread.sleep(2000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            kafkaProducer.close();
        }
    }

    private static String generateRandomName() {
        String[] firstNames = {"Jack", "Jesson", "Alice", "Bob", "Charlie"};
        String[] lastNames = {"Smith", "Johnson", "Williams", "Brown", "Jones"};
        int firstNameIndex = new Random().nextInt(firstNames.length);
        int lastNameIndex = new Random().nextInt(lastNames.length);
        return firstNames[firstNameIndex] + "-" + lastNames[lastNameIndex];
    }

    private static LocalDate generateRandomDate(int startYear, int endYear) {
        LocalDate startDate = LocalDate.of(startYear, 1, 1);
        LocalDate endDate = LocalDate.of(endYear, 12, 31);
        long daysBetween = ChronoUnit.DAYS.between(startDate, endDate);
       Random random = new Random();
        long randomDays = random.nextInt((int) daysBetween);
        return startDate.plusDays(randomDays);
    }

    private static long generateStudentId() {
 Random random = new Random();
    // 生成一个 0 到 8999999999 之间的随机数
    long randomPart = Math.abs(random.nextLong()) % 9000000000L;
    // 确保学号不小于 1000000000
    long studentId = 1000000000L + randomPart;
    // 格式化为8位数字的字符串
    return Long.parseLong(String.format("%08d", studentId));
    }
}